/*
 * Copyright (c) 2015. 石头哥哥
 */

package com.suweia.common.suweiredis.serializations.kryo;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.suweia.common.suweiredis.serializations.Serialization;
import de.javakaffee.kryoserializers.*;
import de.javakaffee.kryoserializers.cglib.CGLibProxySerializer;

import java.lang.reflect.InvocationHandler;
import java.util.Collections;
import java.util.GregorianCalendar;


/**
 * Kryo is not thread safe
 * use thread local
 */
public class KryoThreadLocalSer implements Serialization {


    private KryoThreadLocalSer() {
    }

    /**
     * a singleton of kryo thread local
     *
     * @return
     */

    public static KryoThreadLocalSer getInstance() {
        return Singleton.kryoThreadLocal;
    }


    /**
     * @param obj
     * @return
     */
    @Override
    public byte[] objSerialize(Object obj) {
        KryoHolder kryoHolder = kryoThreadLocal.get();
        kryoHolder.output.clear();  //clear Output    -->每次调用的时候  重置
        kryoHolder.kryo.writeClassAndObject(kryoHolder.output, obj);
        return kryoHolder.output.toBytes();// 无法避免拷贝  ~~~
    }

    /**
     * @param bytes
     * @return
     */
    @Override
    public Object objDeserialize(byte[] bytes) {
        KryoHolder kryoHolder = kryoThreadLocal.get();
        kryoHolder.input.setBuffer(bytes, 0, bytes.length);//call it ,and then use input object  ,discard any array
        return kryoHolder.kryo.readClassAndObject(kryoHolder.input);
    }


    /**
     * creat a Singleton
     */
    private static class Singleton {
        private static final KryoThreadLocalSer kryoThreadLocal = new KryoThreadLocalSer();
    }


    private final ThreadLocal<KryoHolder> kryoThreadLocal = new ThreadLocal<KryoHolder>() {
        @Override
        protected KryoHolder initialValue() {
            return new KryoHolder(new Kryo());
        }
    };


    /**
     * @see KryoPoolSer
     */
    private class KryoHolder {
        private Kryo kryo;
        static final int BUFFER_SIZE = 1024;
        private Output output = new Output(BUFFER_SIZE, -1);     //reuse
        private Input input = new Input();

        KryoHolder(Kryo kryo) {
            this.kryo = kryo;
            this.kryo.setReferences(false);

            //   register
            this.kryo.register(Collections.singletonList("").getClass(), new ArraysAsListSerializer());
            this.kryo.register(Collections.EMPTY_LIST.getClass(), new CollectionsEmptyListSerializer());
            this.kryo.register(Collections.EMPTY_MAP.getClass(), new CollectionsEmptyMapSerializer());
            this.kryo.register(Collections.EMPTY_SET.getClass(), new CollectionsEmptySetSerializer());
            this.kryo.register(Collections.singletonList("").getClass(), new CollectionsSingletonListSerializer());
            this.kryo.register(Collections.singleton("").getClass(), new CollectionsSingletonSetSerializer());
            this.kryo.register(Collections.singletonMap("", "").getClass(), new CollectionsSingletonMapSerializer());
            this.kryo.register(GregorianCalendar.class, new GregorianCalendarSerializer());
            this.kryo.register(InvocationHandler.class, new JdkProxySerializer());
            // register CGLibProxySerializer, works in combination with the appropriate action in handleUnregisteredClass (see below)
            this.kryo.register(CGLibProxySerializer.CGLibProxyMarker.class, new CGLibProxySerializer());
            UnmodifiableCollectionsSerializer.registerSerializers(this.kryo);
            SynchronizedCollectionsSerializer.registerSerializers(this.kryo);

        }

    }


}
